package com.whz.generic.memqueue.disruptor;

import cn.hutool.core.collection.CollUtil;
import com.google.common.collect.Lists;
import com.lmax.disruptor.EventTranslator;
import com.lmax.disruptor.RewindableEventHandler;
import com.lmax.disruptor.TimeoutException;
import com.lmax.disruptor.dsl.Disruptor;
import com.whz.generic.memqueue.MemeQueueProductorHolder;
import com.whz.generic.memqueue.MessageConsumer;
import com.whz.generic.memqueue.SimpleQueue;
import com.whz.generic.memqueue.configure.MemMessageApplicationEvent;
import com.whz.generic.memqueue.configure.MemQueueProperty;
import com.whz.generic.spring.SpringUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.SmartLifecycle;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * disruptor实现的内存队列
 *
 * @author whz
 */
@Slf4j
@Data
public class MemDisruptorQueue
        implements SimpleQueue, SmartLifecycle, RewindableEventHandler<MemEventWrapper> {
    private final Disruptor<MemEventWrapper> disruptor;
    private final MemQueueProperty memQueueProperty;

    /**
     * 事件消费者
     */
    private List<MessageConsumer> messageConsumers = Lists.newArrayList();

    public MemDisruptorQueue(final MemQueueProperty memQueueProperty) {
        this.memQueueProperty = memQueueProperty;
        disruptor = new Disruptor<>(MemEventWrapper::new,
                                    memQueueProperty.getRingBufferSize(),
                                    memQueueProperty.getThreadFactory(),
                                    memQueueProperty.getProducerType(),
                                    memQueueProperty.getWaitStrategy());

        disruptor.handleEventsWith(memQueueProperty.getBatchRewindStrategy(), this);
        //异常处理
        disruptor.setDefaultExceptionHandler(MemQueueProperty.EXCEPTION_HANDLER);
        log.info("初始化内存队列,配置:{}", memQueueProperty);
    }

    @PostConstruct
    public void init() {
        // 初始化完成后,持有这个队列
        MemeQueueProductorHolder.hold(this);
        SpringUtils.getBeans(MessageConsumer.class).ifPresent(consumers -> {
            messageConsumers.addAll(consumers);
        });
        if (CollUtil.isEmpty(messageConsumers)) {
            messageConsumers.add(e -> {
                SpringUtils.getApplicationContext()
                           .ifPresent(app -> app.publishEvent(new MemMessageApplicationEvent(e)));
            });
        }
        log.info("内存队列消费者数量:{}", messageConsumers.size());
    }

    @Override
    public void onEvent(MemEventWrapper event, long sequence, boolean endOfBatch) throws Exception {
        messageConsumers.forEach(consumer -> consumer.consumer(event.getPayload()));
    }

    @Override
    public void start() {
        disruptor.start();
    }

    @Override
    public void stop() {
        try {
            disruptor.shutdown(memQueueProperty.getShutdownTimeout(), memQueueProperty.getShutdownTimeUnit());
        } catch (TimeoutException e) {
            log.error("内存队列关闭超时.", e);
        }
    }

    @Override
    public boolean isRunning() {
        return disruptor.hasStarted();
    }

    @Override
    public void send(Object message) {
        disruptor.publishEvent(new EventTranslator<MemEventWrapper>() {
            @Override
            public void translateTo(MemEventWrapper event, long sequence) {
                event.setPayload(message);
            }
        });
    }

    @Override
    public void addConsumer(MessageConsumer consumer) {
        messageConsumers.add(consumer);
    }
}
